package com.ruoyi.job.config;

import com.ruoyi.common.core.constant.RabbitMqConstants;
import com.ruoyi.system.api.MailServerClient;
import com.ruoyi.system.api.domain.MailSendLog;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.support.atomic.RedisAtomicInteger;

import java.util.concurrent.TimeUnit;

/**
 * @
 */

@Configuration
@Slf4j
public class RabbitMqConfig {

    @Autowired
    RedisTemplate redisTemplate;

    @Autowired
    private ConnectionFactory connectionFactory;

    @Autowired
    MailServerClient mailServerClient;


    @Bean
    public RabbitTemplate rabbitTemplate() {

        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);

        // 必须开启回调才会生效
//        rabbitTemplate.setMandatory(true);

        // 消息确认，需要配置 spring.rabbitmq.publisher-returns = true
        rabbitTemplate.setConfirmCallback((data, ack, cause) -> {
            MailSendLog mailSendLog = new MailSendLog();
            String msgId = data.getId();
            mailSendLog.setMsgId(msgId);

            if (ack) {
                log.debug("消息成功投递到交换机！消息ID: {}", msgId);
            } else {
                log.error("消息投递到交换机失败！消息ID：{}", msgId);
                RedisAtomicInteger redisAtomicInteger = new RedisAtomicInteger(RabbitMqConstants.MAIL_SEND_RETRY_COUNT + msgId, redisTemplate.getConnectionFactory());
                int currentCount = redisAtomicInteger.getAndIncrement();
                //设置个自动过期时间确保不会占用内存
                redisAtomicInteger.expire(3, TimeUnit.MINUTES);
                String userJSON = (String) redisTemplate.opsForValue().get(RabbitMqConstants.CORRELATION_ID + msgId);

                    if (currentCount > 2) {
                        log.info("生产者投递失败，转投失败队列");
                        //重试三次失败就删除计数缓存 发送到失败队列
                        redisTemplate.delete(RabbitMqConstants.MAIL_SEND_RETRY_COUNT + msgId);
                        rabbitTemplate.convertAndSend(RabbitMqConstants.FAILED_EXCHANGE, RabbitMqConstants.FAILED_DELIVER_ROUTING_KEY, true, new CorrelationData(msgId));
                        return;
                    }
                    log.info("生产者投递消息重试中:{}次", currentCount+1);
                    rabbitTemplate.convertAndSend(RabbitMqConstants.RETRY_EXCHANGE, RabbitMqConstants.RETRY_RETRY_ROUTING_KEY, userJSON, new CorrelationData(msgId));
            }
        });

        // 消息发送到交换机但是路由到队列失败  发送到失败队列
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            log.info("ReturnCallback消息发送队列不可达，message:{}，exchange:{}，routingKey:{}，原因:{}", message, exchange, routingKey, replyText);
            String msgId = (String) message.getMessageProperties().getHeaders().get(RabbitMqConstants.MESSAGE_CORRELATION);
            redisTemplate.delete(RabbitMqConstants.MAIL_SEND_RETRY_COUNT+ msgId);
            rabbitTemplate.convertAndSend(RabbitMqConstants.FAILED_EXCHANGE, RabbitMqConstants.FAILED_DELIVER_ROUTING_KEY, true, new CorrelationData(msgId));
        });

        return rabbitTemplate;
    }

}
